package spark.woker

import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import spark.common.{HeartBeat, RegisterWorkerInfo, RegisteredWorkerInfo, SendHeartBeat}

import scala.concurrent.duration._

/**
 * Created by DJJ on 2019/12/2.
 */
class SparkWoker(val serverHost: String, val serverPort: Int, val contextName: String, val serverName: String) extends Actor {

  var masterProxy: ActorSelection = _
  val id = java.util.UUID.randomUUID().toString

  override def preStart(): Unit = {
    println("init-----")
    masterProxy = context.actorSelection(s"akka.tcp://${serverName}@${serverHost}:${serverPort}/user/${contextName}")
    println(masterProxy)
  }

  override def receive: Receive = {
    case "start" => {
      println("worker start")
      masterProxy ! RegisterWorkerInfo(id, 6, 7)
    }
    case RegisteredWorkerInfo => {
      println(s"worker:${id}注册成功")
      import context.dispatcher
      /**
       * 1, 0 millis 发送不延时
       * 2, 3000 millis 每隔3秒执行一次
       * 3,self 发送给自己
       * 4, SendHeartBeat 发送的内容
       */
      context.system.scheduler.schedule(0 millis, 3000 millis, self, SendHeartBeat)
    }
    case SendHeartBeat=>{
      println("worker = " + id + "给master发送心跳")
      masterProxy ! HeartBeat(id)
    }
  }
}

object SparkWoker extends App {

  val (clientHost, clientPort, serverHost, serverPort, contextName, serverName) = ("127.0.0.1", 9990, "127.0.0.1", 10005, "SparkMaster-01", "SparkMaster")
  val config = ConfigFactory.parseString(
    s"""
       |akka.actor.provider="akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname=$clientHost
       |akka.remote.netty.tcp.port=$clientPort
        """.stripMargin)

  val sparkWoker = ActorSystem("SparkWoker", config)
  val woker: ActorRef = sparkWoker.actorOf(Props(new SparkWoker(serverHost, serverPort, contextName, serverName)))
  woker ! "start"
}
